package com.shlh.saas.service.company.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.shlh.saas.common.Desensitization;
import com.shlh.saas.entity.company.Company;
import com.shlh.saas.entity.company.GlobalEnterpriseEs;
import com.shlh.saas.entity.facebook.FacebookUser;
import com.shlh.saas.entity.linkedin.EsLinkedInFields;
import com.shlh.saas.es.EsRestClient;
import com.shlh.saas.es.EsSearchRequest;
import com.shlh.saas.mapper.company.CompanyMapper;
import com.shlh.saas.service.company.CompanyService;
import com.shlh.saas.service.company.GlobalEnterpriseEsService;
import com.shlh.saas.util.CompanyDataSource;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.*;

import org.springframework.jdbc.core.JdbcTemplate;

/**
 * 全球企业库服务实现类
 */
@Service
@Slf4j
public class CompanyServiceImpl extends ServiceImpl<CompanyMapper, Company> implements CompanyService {

    @Autowired
    private CompanyMapper baseMapper;

    @Autowired
    private Desensitization desensitization;

    @Resource
    private JdbcTemplate jdbcTemplate;

    @Resource
    private CompanyDataSource companyDataSource;

    @Resource
    private GlobalEnterpriseEsService globalEnterpriseEsService;

    /**
     * 根据公司名称查询企业
     *
     * @param companyName 公司名称
     * @return 企业列表
     */
    @Override
    public List<Company> findByCompanyName(String companyName) {
        LambdaQueryWrapper<Company> wrapper = new LambdaQueryWrapper<>();
        wrapper.like(Company::getCompanyName, companyName);
        return list(wrapper);
    }

    /**
     * 根据行业查询企业
     *
     * @param industry 行业
     * @return 企业列表
     */
    @Override
    public List<Company> findByIndustry(String industry) {
        LambdaQueryWrapper<Company> wrapper = new LambdaQueryWrapper<>();
        wrapper.like(Company::getIndustry, industry);
        return list(wrapper);
    }

    /**
     * 根据国家查询企业
     *
     * @param country 国家
     * @return 企业列表
     */
    @Override
    public List<Company> findByCountry(String country) {
        LambdaQueryWrapper<Company> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(Company::getCountry, country);
        return list(wrapper);
    }

    /**
     * 根据邮箱查询企业
     *
     * @param email 邮箱
     * @return 企业
     */
    @Override
    public List<Company> findByEmail(String email) {
        List<Map<String, String>> globalEnterpriseEsList = globalEnterpriseEsService.searchGlobalEnterpriseWithEmail(email, 10);

        List<Company> companyList = new ArrayList<>();
        for (Map<String, String> globalEnterpriseEs : globalEnterpriseEsList) {
            String mysqlTable = globalEnterpriseEs.get("mysqlTable");
            String id = globalEnterpriseEs.get("id");
            Company detailByTableAndId = getDetailByTableAndId(id, mysqlTable);
            companyList.add(detailByTableAndId);
        }

        return companyList;
    }

    @Override
    public List<Company> findByPhone(String phone) {
        List<Map<String, String>> globalEnterpriseEsList = globalEnterpriseEsService.searchGlobalEnterpriseWithPhone(phone, 10);

        List<Company> companyList = new ArrayList<>();
        for (Map<String, String> globalEnterpriseEs : globalEnterpriseEsList) {
            String mysqlTable = globalEnterpriseEs.get("mysqlTable");
            String id = globalEnterpriseEs.get("id");
            Company detailByTableAndId = getDetailByTableAndId(id, mysqlTable);
            companyList.add(detailByTableAndId);
        }

        return companyList;
    }


    @Override
    public List<Company> getAllGlobalEnterprises() {
        log.info("获取所有全球企业库数据");
        try {
            return baseMapper.selectAll();
        } catch (Exception e) {
            log.error("获取所有全球企业库数据失败", e);
            return new ArrayList<>();
        }
    }

    /**
     * 分页获取全球企业库数据
     *
     * @param page 页码
     * @param size 每页大小
     * @return 企业列表
     */
    @Override
    public Page<Company> getGlobalEnterprisesPage(int page, int size) {
        log.info("分页获取全球企业库数据，页码：{}，每页大小：{}", page, size);
        try {
//            int offset = (page - 1) * size;
            GlobalEnterpriseEs globalEnterpriseEs = new GlobalEnterpriseEs();
            globalEnterpriseEs.setCurrent(page);
            globalEnterpriseEs.setSize(size);
            Page<GlobalEnterpriseEs> globalEnterpriseEsPage =
                    globalEnterpriseEsService.queryGlobalEnterpriseList(globalEnterpriseEs);

            return covertToCompanyPage(globalEnterpriseEsPage);
        } catch (Exception e) {
            log.error("分页获取全球企业库数据失败", e);
            return new Page<>();
        }
    }

    private Page<Company> covertToCompanyPage(Page<GlobalEnterpriseEs> records) {
        Page<Company> companyPage = new Page<>();
        companyPage.setTotal(records.getTotal());
        companyPage.setSize(records.getSize());
        companyPage.setCurrent(records.getCurrent());
        companyPage.setRecords(getCompanyList(records.getRecords()));
        return companyPage;
    }

    private List<Company> getCompanyList(List<GlobalEnterpriseEs> records) {
        List<Company> companyList = new ArrayList<>();
        for (GlobalEnterpriseEs globalEnterpriseEs : records) {
            Company company = new Company();
            company.setId(globalEnterpriseEs.getId());
            company.setCompanyName(globalEnterpriseEs.getCompanyName());
            company.setFirstName(globalEnterpriseEs.getFirstName());
            company.setLastName(globalEnterpriseEs.getLastName());
            company.setPhone(globalEnterpriseEs.getCompanyPhone());
            company.setMobile(globalEnterpriseEs.getMobilePhone());
            company.setWebAddress(globalEnterpriseEs.getCompanyLinkedInUrl());
            company.setEmail(globalEnterpriseEs.getBusinessEmail());
            company.setIndustry(globalEnterpriseEs.getCompanyIndustry());
            company.setPrimaryState(globalEnterpriseEs.getCompanyState());
            company.setPrimaryCity(globalEnterpriseEs.getCompanyCity());
            company.setMt(globalEnterpriseEs.getMysqlTable());
            company.setMtId(globalEnterpriseEs.getMysqlId());
            company.setEmployee(globalEnterpriseEs.getCompanyEmployeeCount());
            company.setReveneu(globalEnterpriseEs.getCompanyRevenue());
            company.setMailingAddress(globalEnterpriseEs.getCompanyAddress());
            company.setTitle(globalEnterpriseEs.getJobTitle());
            company.setSubIndustry(globalEnterpriseEs.getPrimaryIndustry());
            company.setZipCode(globalEnterpriseEs.getCompanyZip());
            company.setDepartment(globalEnterpriseEs.getDepartment());
            companyList.add(company);
        }
        CompaniesDesensitization(companyList);
        return companyList;
    }


    /**
     * 统计全球企业库数据总数
     *
     * @return 数据总数
     */
    @Override
    public long countGlobalEnterprises() {
        log.info("统计全球企业库数据总数");
        try {
            GlobalEnterpriseEs globalEnterpriseEs = new GlobalEnterpriseEs();
            return globalEnterpriseEsService.countGlobalEnterprise(globalEnterpriseEs);
        } catch (Exception e) {
            log.error("统计全球企业库数据总数失败", e);
            return 0L;
        }
    }

    @Override
    public List<Company> searchGlobalEnterprises(String companyName, String country, String city, String industry, String employee, String reveneu) {
        log.info("搜索全球企业库数据，参数：companyName={}, country={}, city={}, industry={}, employee={}, reveneu={}",
                companyName, country, city, industry, employee, reveneu);
        try {
            Map<String, Object> params = new HashMap<>();
            if (StringUtils.hasText(companyName)) {
                params.put("companyName", "%" + companyName + "%");
            }
            if (StringUtils.hasText(country)) {
                params.put("country", country);
            }
            if (StringUtils.hasText(city)) {
                params.put("city", city);
            }
            if (StringUtils.hasText(industry)) {
                params.put("industry", industry);
            }
            if (StringUtils.hasText(employee)) {
                params.put("employee", employee);
            }
            if (StringUtils.hasText(reveneu)) {
                params.put("reveneu", reveneu);
            }
            return baseMapper.searchByCondition(params);
        } catch (Exception e) {
            log.error("搜索全球企业库数据失败", e);
            return new ArrayList<>();
        }
    }

    /**
     * 根据条件分页搜索全球企业库数据
     *
     * @param companyName 公司名称
     * @param country     国家
     * @param city        城市
     * @param industry    行业
     * @param employee    员工规模
     * @param reveneu     收入规模
     * @param page        页码
     * @param size        每页大小
     * @return 符合条件的企业列表
     */
    @Override
    public Page<Company> searchGlobalEnterprisesPage(String companyName, String firstName, String lastName, String country, String city, String industry,
                                                     String employee, String reveneu, String zipcode, String department, String occupation, String state, String hasContact, int page, int size) {
        log.info("分页搜索全球企业库数据，参数：companyName={}, firstName={}, lastName={}, country={}, city={}, industry={}, employee={}, reveneu={}, zipcode={}, department={}, occupation={}, state={}, hasContact={}, page={}, size={}",
                companyName, firstName, lastName, country, city, industry, employee, reveneu, zipcode, department, occupation, state, hasContact, page, size);
        try {
            GlobalEnterpriseEs globalEnterpriseEs = new GlobalEnterpriseEs();
            if (StringUtils.hasText(companyName)) {
                globalEnterpriseEs.setCompanyName(companyName);
            }
            if (StringUtils.hasText(firstName)) {
                globalEnterpriseEs.setFirstName(firstName);
            }
            if (StringUtils.hasText(lastName)) {
                globalEnterpriseEs.setLastName(lastName);
            }
            if (StringUtils.hasText(city)) {
                globalEnterpriseEs.setCompanyCity(city);
            }
            if (StringUtils.hasText(industry)) {
                globalEnterpriseEs.setCompanyIndustry(industry);
            }
            if (StringUtils.hasText(employee)) {
                globalEnterpriseEs.setEmployeeSizeCategory(employee);
            }
            if (StringUtils.hasText(reveneu)) {
                globalEnterpriseEs.setRevenueSizeCategory(reveneu);
            }
            if (StringUtils.hasText(zipcode)) {
                globalEnterpriseEs.setCompanyZip(zipcode);
            }
            if (StringUtils.hasText(department)) {
                globalEnterpriseEs.setDepartment(department);
            }
            if (StringUtils.hasText(occupation)) {
                globalEnterpriseEs.setJobTitle(occupation);
            }
            if (StringUtils.hasText(state)) {
                globalEnterpriseEs.setCompanyState(state);
            }
            if (StringUtils.hasText(hasContact)) {
                globalEnterpriseEs.setHasContact(hasContact);
            }
            globalEnterpriseEs.setCurrent(page);
            globalEnterpriseEs.setSize(size);
            Page<Company> companyPage =
                    covertToCompanyPage(globalEnterpriseEsService.queryGlobalEnterpriseList(globalEnterpriseEs));
            return companyPage;
        } catch (Exception e) {
            log.error("分页搜索全球企业库数据失败", e);
            return new Page<>();
        }
    }

    @NotNull
    private List<Company> CompaniesDesensitization(List<Company> companyList) {
        for (Company company : companyList) {
            if (StringUtils.hasText(company.getEmail())) {
                // 加密
                String email = desensitization.maskEmail(company.getEmail());
                company.setEmail(email);
            }
            if (StringUtils.hasText(company.getPhone())) {
                // 加密
                String phone = desensitization.maskPhone(company.getPhone());
                company.setPhone(phone);
            }
            if (StringUtils.hasText(company.getMobile())) {
                // 加密
                String mobile = desensitization.maskPhone(company.getMobile());
                company.setMobile(mobile);
            }
            if (StringUtils.hasText(company.getWebAddress())) {
                // 加密
                String webAddress = desensitization.maskWebAddress(company.getWebAddress());
                company.setWebAddress(webAddress);
            }
        }
        return companyList;
    }

    /**
     * 统计符合条件的全球企业库数据总数
     *
     * @param companyName 公司名称
     * @param country     国家
     * @param city        城市
     * @param industry    行业
     * @param employee    员工规模
     * @param reveneu     收入规模
     * @return 数据总数
     */
    @Override
    public long countSearchGlobalEnterprises(String companyName, String country, String city, String industry, String employee, String reveneu,
                                             String firstName, String lastName, String zipcode, String department, String occupation, String state, String hasContact) {
        log.info("统计符合条件的全球企业库数据总数，参数：companyName={}, country={}, city={}, industry={}, employee={}, reveneu={}, zipcode={}, department={}, occupation={}, state={}, hasContact={}",
                companyName, country, city, industry, employee, reveneu, zipcode, department, occupation, state, hasContact);
        try {
            GlobalEnterpriseEs globalEnterpriseEs = new GlobalEnterpriseEs();
            if (StringUtils.hasText(companyName)) {
                globalEnterpriseEs.setCompanyName(companyName);
            }
            if (StringUtils.hasText(firstName)) {
                globalEnterpriseEs.setFirstName(firstName);
            }
            if (StringUtils.hasText(lastName)) {
                globalEnterpriseEs.setLastName(lastName);
            }
            if (StringUtils.hasText(city)) {
                globalEnterpriseEs.setCompanyCity(city);
            }
            if (StringUtils.hasText(industry)) {
                globalEnterpriseEs.setCompanyIndustry(industry);
            }
            if (StringUtils.hasText(employee)) {
                globalEnterpriseEs.setEmployeeSizeCategory(employee);
            }
            if (StringUtils.hasText(reveneu)) {
                globalEnterpriseEs.setRevenueSizeCategory(reveneu);
            }
            if (StringUtils.hasText(zipcode)) {
                globalEnterpriseEs.setCompanyZip(zipcode);
            }
            if (StringUtils.hasText(department)) {
                globalEnterpriseEs.setDepartment(department);
            }
            if (StringUtils.hasText(occupation)) {
                globalEnterpriseEs.setJobTitle(occupation);
            }
            if (StringUtils.hasText(state)) {
                globalEnterpriseEs.setCompanyState(state);
            }
            if (StringUtils.hasText(hasContact)) {
                globalEnterpriseEs.setHasContact(hasContact);
            }

            return globalEnterpriseEsService.countGlobalEnterprise(globalEnterpriseEs);
        } catch (Exception e) {
            log.error("统计符合条件的全球企业库数据总数失败", e);
            return 0L;
        }
    }

    @Override
    public List<Company> searchByKeywordInAllFields(String keyword, int offset, int limit) {
        log.info("在所有字段中搜索关键词：keyword={}, offset={}, limit={}", keyword, offset, limit);
        try {
            return baseMapper.searchInAllFieldsWithPagination(keyword, offset, limit);
        } catch (Exception e) {
            log.error("在所有字段中搜索关键词失败", e);
            return new ArrayList<>();
        }
    }

    @Override
    public long countByKeywordInAllFields(String keyword) {
        log.info("统计包含关键词的记录数量：keyword={}", keyword);
        try {
            return baseMapper.countInAllFields(keyword);
        } catch (Exception e) {
            log.error("统计包含关键词的记录数量失败", e);
            return 0L;
        }
    }

    @Override
    public List<String> getSuggestions(String keyword) {
        log.info("获取搜索建议：keyword={}", keyword);
        try {
            List<String> suggestions = new ArrayList<>();
            suggestions.addAll(baseMapper.findSimilarCompanyNames(keyword));
            suggestions.addAll(baseMapper.findSimilarIndustries(keyword));
            return suggestions;
        } catch (Exception e) {
            log.error("获取搜索建议失败", e);
            return new ArrayList<>();
        }
    }

    @Override
    public List<Company> searchGlobalEnterprisesForExport(Map<String, Object> params) {
        log.info("搜索要导出的全球企业数据（两阶段查询模式），参数: {}", params);

        try {
            // 第一阶段：从ES获取企业标识信息
            List<Map<String, Object>> identifiers = getEnterpriseIdentifiersFromES(params);
            log.info("第一阶段ES查询完成，获取到 {} 条标识信息", identifiers.size());

            if (identifiers.isEmpty()) {
                log.info("ES查询未找到符合条件的数据");
                return new ArrayList<>();
            }

            // 第二阶段：根据标识信息从数据库查询完整数据
            List<Company> completeData = getCompleteEnterpriseDataFromDB(identifiers);
            log.info("第二阶段数据库查询完成，获取到 {} 条完整数据", completeData.size());

            return completeData;

        } catch (Exception e) {
            log.error("两阶段查询导出数据时发生错误", e);
            return new ArrayList<>();
        }
    }

    @Override
    public List<Map<String, Object>> getEnterpriseIdentifiersFromES(Map<String, Object> params) {
        log.info("第一阶段：从ES查询企业标识信息，参数: {}", params);

        try {
            // 构建ES查询条件
            GlobalEnterpriseEs globalEnterpriseEs = new GlobalEnterpriseEs();

            // 解析查询参数
            String companyName = params != null && params.containsKey("companyName") ?
                    (String) params.get("companyName") : null;
            String firstName = params != null && params.containsKey("firstName") ?
                    (String) params.get("firstName") : null;
            String lastName = params != null && params.containsKey("lastName") ?
                    (String) params.get("lastName") : null;
            String zipcode = params != null && params.containsKey("zipcode") ?
                    (String) params.get("zipcode") : null;
            String state = params != null && params.containsKey("state") ?
                    (String) params.get("state") : null;
            String hasContact = params != null && params.containsKey("hasContact") ?
                    (String) params.get("hasContact") : null;

            List<String> countries = params != null && params.containsKey("country") ?
                    (List<String>) params.get("country") : null;
            List<String> cities = params != null && params.containsKey("primaryCity") ?
                    (List<String>) params.get("primaryCity") : null;
            List<String> industries = params != null && params.containsKey("industry") ?
                    (List<String>) params.get("industry") : null;
            List<String> employees = params != null && params.containsKey("employee") ?
                    (List<String>) params.get("employee") : null;
            List<String> reveneus = params != null && params.containsKey("reveneu") ?
                    (List<String>) params.get("reveneu") : null;
            List<String> departments = params != null && params.containsKey("department") ?
                    (List<String>) params.get("department") : null;
            List<String> occupations = params != null && params.containsKey("occupation") ?
                    (List<String>) params.get("occupation") : null;


            // 设置查询条件
            if (StringUtils.hasText(companyName)) {
                globalEnterpriseEs.setCompanyName(companyName);
            }
            if (StringUtils.hasText(firstName)) {
                globalEnterpriseEs.setFirstName(firstName);
            }
            if (StringUtils.hasText(lastName)) {
                globalEnterpriseEs.setLastName(lastName);
            }
            if (StringUtils.hasText(zipcode)) {
                globalEnterpriseEs.setCompanyZip(zipcode);
            }
            if (StringUtils.hasText(state)) {
                globalEnterpriseEs.setCompanyState(state);
            }
//            if (countries != null && !countries.isEmpty()) {
//                globalEnterpriseEs.setCountry(countries.get(0));
//            }
            if (cities != null && !cities.isEmpty()) {
                globalEnterpriseEs.setCompanyCity(cities.get(0));
            }
            if (industries != null && !industries.isEmpty()) {
                globalEnterpriseEs.setCompanyIndustry(industries.get(0));
            }
            if (employees != null && !employees.isEmpty()) {
                globalEnterpriseEs.setEmployeeSizeCategory(employees.get(0));
            }
            if (reveneus != null && !reveneus.isEmpty()) {
                globalEnterpriseEs.setRevenueSizeCategory(reveneus.get(0));
            }
            if (departments != null && !departments.isEmpty()) {
                globalEnterpriseEs.setDepartment(departments.get(0));
            }
            if (occupations != null && !occupations.isEmpty()) {
                globalEnterpriseEs.setJobTitle(occupations.get(0));
            }
            if (hasContact != null && !hasContact.isEmpty()) {
                globalEnterpriseEs.setHasContact(hasContact);
            }

            // 获取导出数量限制
            Integer limit = params != null && params.containsKey("limit") ?
                    (Integer) params.get("limit") : 5000;

            // 设置分页参数（用于导出，设置较大的页面大小）
            globalEnterpriseEs.setCurrent(1);
            globalEnterpriseEs.setSize(limit);

            // 执行ES查询
            Page<GlobalEnterpriseEs> esPage = globalEnterpriseEsService.queryGlobalEnterpriseList(globalEnterpriseEs);

            // 提取标识信息
            List<Map<String, Object>> identifiers = new ArrayList<>();
            for (GlobalEnterpriseEs esRecord : esPage.getRecords()) {
                Map<String, Object> identifier = new HashMap<>();
                identifier.put("tableName", esRecord.getMysqlTable());
                identifier.put("recordId", esRecord.getMysqlId());
                // 保留一些基本信息用于日志和调试
                identifier.put("companyName", esRecord.getCompanyName());
                identifiers.add(identifier);
            }

            log.info("ES查询返回 {} 条标识信息", identifiers.size());
            return identifiers;

        } catch (Exception e) {
            log.error("从ES查询企业标识信息时发生错误", e);
            return new ArrayList<>();
        }
    }

    @Override
    public List<Company> getCompleteEnterpriseDataFromDB(List<Map<String, Object>> identifiers) {
        log.info("第二阶段：根据标识信息从数据库查询完整企业数据，标识数量: {}", identifiers.size());

        if (identifiers == null || identifiers.isEmpty()) {
            return new ArrayList<>();
        }

        List<Company> completeData = new ArrayList<>();

        try {
            // 按表名分组，批量查询提高效率
            Map<String, List<String>> tableGroups = new HashMap<>();

            for (Map<String, Object> identifier : identifiers) {
                String tableName = (String) identifier.get("tableName");
                String recordId = (String) identifier.get("recordId");

                if (StringUtils.hasText(tableName) && StringUtils.hasText(recordId)) {
                    tableGroups.computeIfAbsent(tableName, k -> new ArrayList<>()).add(recordId);
                }
            }

            log.info("按表名分组后，需要查询 {} 个表", tableGroups.size());

            // 遍历每个表，批量查询数据
            for (Map.Entry<String, List<String>> entry : tableGroups.entrySet()) {
                String tableName = entry.getKey();
                List<String> recordIds = entry.getValue();

                log.info("查询表 {} 中的 {} 条记录", tableName, recordIds.size());

                // 根据表名和ID列表查询完整数据
                List<Company> tableData = queryCompleteDataByTableAndIds(tableName, recordIds);
                completeData.addAll(tableData);

                log.info("从表 {} 查询到 {} 条完整数据", tableName, tableData.size());
            }

            log.info("第二阶段查询完成，总共获取到 {} 条完整数据", completeData.size());

            // 对数据进行脱敏处理
//            CompaniesDesensitization(completeData);

            return completeData;

        } catch (Exception e) {
            log.error("从数据库查询完整企业数据时发生错误", e);
            return new ArrayList<>();
        }
    }

    /**
     * 根据表名和ID列表查询完整的企业数据
     *
     * @param tableName 表名
     * @param recordIds 记录ID列表
     * @return 完整的企业数据列表
     */
    private List<Company> queryCompleteDataByTableAndIds(String tableName, List<String> recordIds) {
        log.info("批量查询表 {} 中ID为 {} 的记录", tableName, recordIds);

        try {
            // 优先使用批量查询提高效率
            if (recordIds.size() > 1) {
                log.info("使用批量查询模式，查询 {} 条记录", recordIds.size());
                List<Company> results = baseMapper.batchQueryByTableAndIds(tableName, recordIds);
                log.info("批量查询返回 {} 条记录", results.size());
                return results;
            } else if (recordIds.size() == 1) {
                // 单条记录使用现有的单个查询方法
                log.info("使用单个查询模式");
                Company company = getDetailByTableAndId(recordIds.get(0), tableName);
                List<Company> results = new ArrayList<>();
                if (company != null) {
                    results.add(company);
                }
                return results;
            } else {
                return new ArrayList<>();
            }

        } catch (Exception e) {
            log.error("批量查询表 {} 中的记录时发生错误，回退到逐个查询模式", tableName, e);

            // 如果批量查询失败，回退到逐个查询模式
            List<Company> results = new ArrayList<>();
            for (String recordId : recordIds) {
                try {
                    Company company = getDetailByTableAndId(recordId, tableName);
                    if (company != null) {
                        results.add(company);
                    }
                } catch (Exception ex) {
                    log.warn("查询表 {} 中ID {} 的记录时出错: {}", tableName, recordId, ex.getMessage());
                    // 继续处理其他记录，不因单个记录错误而中断整个查询
                }
            }
            return results;
        }
    }

    /**
     * 根据ID查询企业
     *
     * @param mysqlTable 企业ID
     * @return 企业信息
     */
    @Override
    public Company getDetailByTableAndId(String id, String mysqlTable) {
        log.info("根据ID查询企业: {}", id);
        try {
            log.info("根据数据表和ID查询企业: mysqlTable={}, id={}", mysqlTable, id);

            // 参数验证
            if (!StringUtils.hasText(mysqlTable) || !StringUtils.hasText(id)) {
                log.warn("参数不能为空: mysqlTable={}, id={}", mysqlTable, id);
                return null;
            }

            // 验证数据表名是否安全（防止SQL注入）
            if (!isValidTableName(mysqlTable)) {
                log.warn("无效的数据表名: {}", mysqlTable);
                return null;
            }
            // 查询企业信息

            // 使用ShardingSphere兼容的查询方式
            List<Map<String, Object>> results = executeShardingSphereCompatibleQuery(mysqlTable, id);

            if (results.isEmpty()) {
                log.info("未查询到企业信息: table={}, id={}", mysqlTable, id);
                return null;
            }

            // 将查询结果转换为FacebookUser对象
            Map<String, Object> row = results.get(0);
            Company company = convertMapToCompany(row);

            log.info("成功获取联系人详情: {}",
                    JSON.toJSONString(company));

            return company;


        } catch (Exception e) {
            log.error("根据ID查询企业失败", e);
            return null;
        }
    }

    /**
     * 批量获取Company详情（按表名和ID列表）
     *
     * @param mysqlTable 数据表名
     * @param ids        ID列表
     * @return Company列表
     */
    private List<Company> batchGetCompanyDetailByTableAndIds(String mysqlTable, List<String> ids) {
        long startTime = System.currentTimeMillis();
        log.info("批量查询表{}: 包含{}条记录", mysqlTable, ids.size());

        try {
            // 验证表名安全
            if (!isValidTableName(mysqlTable)) {
                log.warn("无效的数据表名: {}", mysqlTable);
                return new ArrayList<>();
            }

            if (ids == null || ids.isEmpty()) {
                return new ArrayList<>();
            }

            // 分批查询（每批最多1000条，避免SQL过长）
            int batchSize = 1000;
            List<Company> allResults = new ArrayList<>();

            for (int i = 0; i < ids.size(); i += batchSize) {
                int end = Math.min(i + batchSize, ids.size());
                List<String> batchIds = ids.subList(i, end);

                // 构建IN查询的占位符
                String placeholders = String.join(",", Collections.nCopies(batchIds.size(), "?"));
                String sql = String.format(
                        "/* ShardingSphere hint: dataSourceName=zoominfo */ SELECT * FROM zoominfo.%s WHERE id IN (%s)",
                        mysqlTable, placeholders
                );

                log.debug("执行批量查询: 表={}, 条数={}, SQL长度={}", mysqlTable, batchIds.size(), sql.length());

                // 执行查询
                jdbcTemplate.setDataSource(companyDataSource.dataSource());
                List<Map<String, Object>> results = jdbcTemplate.queryForList(sql, batchIds.toArray());

                // 转换结果
                for (Map<String, Object> row : results) {
                    Company company = convertMapToCompany(row);
                    if (company != null) {
                        allResults.add(company);
                    }
                }
            }

            long endTime = System.currentTimeMillis();
            log.info("批量查询表{}完成: 请求{}条, 成功{}条, 耗时{}ms",
                    mysqlTable, ids.size(), allResults.size(), (endTime - startTime));

            return allResults;

        } catch (Exception e) {
            log.error("批量查询表{}失败: ids数量={}", mysqlTable, ids.size(), e);
            return new ArrayList<>();
        }
    }

    @Override
    public List<Company> searchCompaniesWithEmail(String searchKeyword, int offset, int size) {
        List<GlobalEnterpriseEs> globalEnterpriseEsList = globalEnterpriseEsService.searchGlobalEnterpriseByEmail(searchKeyword, offset, size);
        List<Company> companyList = new ArrayList<>();
        for (GlobalEnterpriseEs globalEnterpriseEs : globalEnterpriseEsList) {
            Company company;
            String mysqlTable = globalEnterpriseEs.getMysqlTable();
            String id = globalEnterpriseEs.getMysqlId();
            company = getDetailByTableAndId(id, mysqlTable);
            companyList.add(company);
        }

        return companyList;

    }

    @Override
    public Long countCompaniesWithEmail(String searchKeyword) {
        return globalEnterpriseEsService.countGlobalEnterpriseByEmail(searchKeyword);
    }

    @Override
    public List<Company> searchCompaniesWithPhone(String searchKeyword, int offset, int size) {
        List<GlobalEnterpriseEs> globalEnterpriseEsList = globalEnterpriseEsService.searchCompaniesWithPhone(searchKeyword, offset, size);
        List<Company> companyList = new ArrayList<>();
        for (GlobalEnterpriseEs globalEnterpriseEs : globalEnterpriseEsList) {
            Company company;
            String mysqlTable = globalEnterpriseEs.getMysqlTable();
            String id = globalEnterpriseEs.getMysqlId();
            company = getDetailByTableAndId(id, mysqlTable);
            companyList.add(company);
        }

        return companyList;
    }

    @Override
    public Long countCompaniesWithPhone(String searchKeyword) {
        return globalEnterpriseEsService.countCompaniesWithPhone(searchKeyword);
    }

    @Override
    public List<Map<String, Object>> searchByEmail(String value, int size, int offset) {
        List<Map<String, String>> globalEnterpriseEsList = globalEnterpriseEsService.findGlobalEnterpriseByEmail(value, offset, size);

        return batchQueryCompanyFromMysql(globalEnterpriseEsList);
        //        List<Map<String, Object>> result = new ArrayList<>();
//        for (Map<String, String> globalEnterpriseEs : globalEnterpriseEsList) {
//            Map<String, Object> companyMap = getByTableAndId(globalEnterpriseEs.get("id"), globalEnterpriseEs.get("mysqlTable"));
//            result.add(companyMap);
//        }
//        return result;
    }

    @Override
    public List<Map<String, Object>> searchByPhone(String value, int size, int offset) {
        List<Map<String, String>> globalEnterpriseEsList = globalEnterpriseEsService.findGlobalEnterpriseByPhone(value, offset, size);

        return batchQueryCompanyFromMysql(globalEnterpriseEsList);
//        List<Map<String, Object>> result = new ArrayList<>();
//        for (Map<String, String> globalEnterpriseEs : globalEnterpriseEsList) {
//            Map<String, Object> companyMap = getByTableAndId(globalEnterpriseEs.get("id"), globalEnterpriseEs.get("mysqlTable"));
//            result.add(companyMap);
//        }
//        return result;
    }

    @Override
    public long countByEmail(String value) {
        GlobalEnterpriseEs globalEnterpriseEs = new GlobalEnterpriseEs();
        globalEnterpriseEs.setBusinessEmail(value);
        return globalEnterpriseEsService.countGlobalEnterprise(globalEnterpriseEs);
    }

    @Override
    public long countByPhone(String value) {
        GlobalEnterpriseEs globalEnterpriseEs = new GlobalEnterpriseEs();
        globalEnterpriseEs.setCompanyPhone(value);
        return globalEnterpriseEsService.countGlobalEnterprise(globalEnterpriseEs);
    }

    @Override
    public List<Map<String, Object>> findByDomain(String domain, int size, int offset) {
        log.info("通过域名查询全球企业库: domain={}, size={}, offset={}", domain, size, offset);

        try {
            // 第一阶段：从ES获取标识信息
            List<Map<String, String>> identifiers = globalEnterpriseEsService.findGlobalEnterpriseByDomain(domain, size, offset);

            if (identifiers.isEmpty()) {
                log.info("ES查询未找到匹配数据");
                return new ArrayList<>();
            }

            log.info("ES查询完成，获取到{}条标识信息", identifiers.size());

            // 第二阶段：批量从数据库查询完整数据
            return batchQueryCompanyFromMysql(identifiers);
        } catch (Exception e) {
            log.error("通过域名查询全球企业库失败: domain={}", domain, e);
            return new ArrayList<>();
        }
    }

    /**
     * 批量从MySQL查询全球企业库数据
     */
    private List<Map<String, Object>> batchQueryCompanyFromMysql(List<Map<String, String>> identifiers) {
        List<Map<String, Object>> results = new ArrayList<>();
        jdbcTemplate.setDataSource(companyDataSource.dataSource());

        try {
            // 按表名分组
            Map<String, List<String>> tableIdMap = new HashMap<>();
            for (Map<String, String> identifier : identifiers) {
                String table = identifier.get("mysqlTable");
                String id = identifier.get("id");

                if (table != null && id != null) {
                    tableIdMap.computeIfAbsent(table, k -> new ArrayList<>()).add(id);
                }
            }

            // 对每个表进行批量ID查询
            for (Map.Entry<String, List<String>> entry : tableIdMap.entrySet()) {
                String tableName = entry.getKey();
                List<String> ids = entry.getValue();

                if (ids.isEmpty()) {
                    continue;
                }

                try {
                    // 批量查询 - 使用ShardingSphere Hint强制指定数据源
                    String placeholders = String.join(",", ids.stream().map(id -> "?").toList());
                    String sql = String.format(
                            "/* ShardingSphere hint: dataSourceName=zoominfo */ SELECT * FROM zoominfo.%s WHERE id IN (%s)",
                            tableName, placeholders
                    );

                    Object[] params = ids.toArray();
                    List<Map<String, Object>> tableResults = jdbcTemplate.queryForList(sql, params);
                    results.addAll(tableResults);

                    log.info("从表{}批量查询{}条记录，实际查询到{}条", tableName, ids.size(), tableResults.size());
                } catch (Exception e) {
                    log.error("批量查询表{}失败: ", tableName, e);
                }
            }

            return results;
        } catch (Exception e) {
            log.error("批量查询全球企业库数据失败", e);
            return new ArrayList<>();
        }
    }

    @Override
    public long countByDomain(String domain) {
        GlobalEnterpriseEs globalEnterpriseEs = new GlobalEnterpriseEs();
        globalEnterpriseEs.setCompanyDomain(domain);
        return globalEnterpriseEsService.countGlobalEnterprise(globalEnterpriseEs);
    }

    @Override
    public List<String> searchCities(String cities) {
        return globalEnterpriseEsService.searchCities(cities);
    }

    @Override
    public List<String> searchIndustries(String industry) {
        return globalEnterpriseEsService.searchIndustries(industry);
    }

    @Override
    public List<String> searchOccupations(String occupations) {
        return globalEnterpriseEsService.searchOccupations(occupations);
    }

    @Override
    public List<String> searchDepartments(String departments) {
        return globalEnterpriseEsService.searchDepartments(departments);
    }

    @Override
    public List<Company> searchCompanyData(String companyName, String fullName, String website, Integer matchType) {
        log.info("批量搜索Company: 公司名称={}, 人名={}, 网址={}, 匹配类型={}", companyName, fullName, website, matchType);

        try {
            List<GlobalEnterpriseEs> companyList;
            if (matchType == 1) {
                companyList = globalEnterpriseEsService.searchCompanyDataByPhone(companyName, fullName, website);
            } else {
                companyList = globalEnterpriseEsService.searchCompanyDataByEmail(companyName, fullName, website);
            }

            log.info("ES查询到{}条Company数据，开始从MySQL批量获取详细信息", companyList.size());

            if (companyList.isEmpty()) {
                return new ArrayList<>();
            }

            // 按表名分组
            Map<String, List<String>> tableToIds = new HashMap<>();
            for (GlobalEnterpriseEs globalEnterpriseEs : companyList) {
                String table = globalEnterpriseEs.getMysqlTable();
                String id = globalEnterpriseEs.getMysqlId();
                if (table != null && id != null) {
                    tableToIds.computeIfAbsent(table, k -> new ArrayList<>()).add(id);
                }
            }

            log.info("数据分布在{}个不同的表中，开始并行批量查询", tableToIds.size());

            // 使用线程池并行查询不同的表
            ExecutorService queryExecutor = Executors.newFixedThreadPool(Math.min(tableToIds.size(), 20));
            List<CompletableFuture<List<Company>>> queryTasks = new ArrayList<>();

            for (Map.Entry<String, List<String>> entry : tableToIds.entrySet()) {
                String table = entry.getKey();
                List<String> ids = entry.getValue();

                CompletableFuture<List<Company>> task = CompletableFuture.supplyAsync(() -> {
                    try {
                        return batchGetCompanyDetailByTableAndIds(table, ids);
                    } catch (Exception e) {
                        log.error("批量查询表{}失败，包含{}条记录", table, ids.size(), e);
                        return new ArrayList<>();
                    }
                }, queryExecutor);

                queryTasks.add(task);
            }

            // 等待所有查询完成
            CompletableFuture<Void> allTasks = CompletableFuture.allOf(
                    queryTasks.toArray(new CompletableFuture[0])
            );
            allTasks.join();

            // 合并结果
            List<Company> result = new ArrayList<>();
            for (CompletableFuture<List<Company>> task : queryTasks) {
                try {
                    List<Company> results = task.get();
                    result.addAll(results);
                } catch (Exception e) {
                    log.error("获取查询结果失败", e);
                }
            }

            // 关闭线程池
            queryExecutor.shutdown();

            int successCount = result.size();
            int failCount = companyList.size() - successCount;

            log.info("Company数据处理完成: ES查询{}条, MySQL成功获取{}条, 失败{}条",
                    companyList.size(), successCount, failCount);
            return result;

        } catch (Exception e) {
            log.error("Company库查询失败", e);
            return new ArrayList<>();
        }
    }

    private Map<String, Object> getByTableAndId(String id, String mysqlTable) {
        log.info("根据ID查询企业: {}", id);
        try {
            log.info("根据数据表和ID查询企业: mysqlTable={}, id={}", mysqlTable, id);

            // 参数验证
            if (!StringUtils.hasText(mysqlTable) || !StringUtils.hasText(id)) {
                log.warn("参数不能为空: mysqlTable={}, id={}", mysqlTable, id);
                return null;
            }

            // 验证数据表名是否安全（防止SQL注入）
            if (!isValidTableName(mysqlTable)) {
                log.warn("无效的数据表名: {}", mysqlTable);
                return null;
            }
            // 查询企业信息

            // 使用ShardingSphere兼容的查询方式
            List<Map<String, Object>> results = executeShardingSphereCompatibleQuery(mysqlTable, id);

            if (results.isEmpty()) {
                log.info("未查询到企业信息: table={}, id={}", mysqlTable, id);
                return null;
            }

            return results.get(0);

        } catch (Exception e) {
            log.error("根据ID查询企业失败", e);
            return null;
        }
    }

    private Company convertMapToCompany(Map<String, Object> row) {
        Company company = new Company();

        company.setCompanyName(getString(row, "COMPANY_NAME"));
        company.setCountry(getString(row, "COUNTRY"));
        company.setFirstName(getString(row, "FIRST_NAME"));
        company.setLastName(getString(row, "LAST_NAME"));
        company.setPhone(getString(row, "COMPANY_PHONE"));
        company.setMobile(getString(row, "MOBILE_PHONE"));
        company.setWebAddress(getString(row, "COMPANY_LINKEDIN_URL"));
        company.setEmail(getString(row, "BUSINESS_EMAIL"));
        company.setIndustry(getString(row, "COMPANY_INDUSTRY"));
        company.setPrimaryState(getString(row, "COMPANY_STATE"));
        company.setPrimaryCity(getString(row, "COMPANY_CITY"));
        company.setId((Long) row.get("id"));
        company.setEmployee(getString(row, "COMPANY_EMPLOYEE_COUNT"));
        company.setReveneu(getString(row, "COMPANY_REVENUE"));
        company.setMailingAddress(getString(row, "COMPANY_ADDRESS"));
        company.setTitle(getString(row, "JOB_TITLE"));
        company.setSubIndustry(getString(row, "PRIMARY_INDUSTRY"));
        company.setZipCode(getString(row, "COMPANY_ZIP"));
        company.setCompanyDomain(getString(row, "COMPANY_DOMAIN"));

        return company;

    }

    /**
     * 安全地从Map中获取字符串值
     *
     * @param row 数据行
     * @param key 键名
     * @return 字符串值
     */
    private String getString(Map<String, Object> row, String key) {
        Object value = row.get(key);
        return value != null ? value.toString() : null;
    }

    private List<Map<String, Object>> executeShardingSphereCompatibleQuery(String mysqlTable, String id) {
        try {
            // 使用ShardingSphere兼容的查询方式
            jdbcTemplate.setDataSource(companyDataSource.dataSource());

            // 使用ShardingSphere Hint强制指定数据源
            String sql =
                    "/* ShardingSphere hint: dataSourceName=zoominfo */ SELECT * FROM zoominfo." + mysqlTable + " WHERE id = ?";
            log.debug("执行备用查询: {}, 参数: {}", sql, id);

            return jdbcTemplate.queryForList(sql, id);

        } catch (Exception e) {
            log.error("执行备用查询失败", e);
            return new ArrayList<>();
        }
    }


    private boolean isValidTableName(String tableName) {
        // 识别表名p001到p369
        return tableName.matches("^p(?:00[1-9]|0[1-9]\\d|[1-2]\\d{2}|3[0-6]\\d)$");
    }

    /**
     * 使用多线程批量处理方式查询全球企业数据（针对大数据量导出优化）
     *
     * @param params 查询参数
     * @return 企业列表
     */
    @Override
    public List<Company> searchGlobalEnterprisesForExportOptimized(Map<String, Object> params) {
        log.info("优化查询全球企业数据，使用多线程批处理方式");

        // 第一阶段：从ES获取标识信息
        log.info("第一阶段：从ES获取企业标识信息，参数: {}", params);

        // 获取标识信息
        List<Map<String, Object>> identifiers = getEnterpriseIdentifiersFromES(params);
        log.info("第一阶段ES查询完成，获取到 {} 条标识信息", identifiers.size());

        if (identifiers.isEmpty()) {
            log.info("ES查询未找到符合条件的数据");
            return new ArrayList<>();
        }

        // 第二阶段：按表名分组后，使用多线程并行查询数据库
        log.info("第二阶段：使用多线程批量查询数据库");

        // 1. 按表名分组
        Map<String, List<String>> groupedByTable = new HashMap<>();
        for (Map<String, Object> identifier : identifiers) {
            String tableName = (String) identifier.get("tableName");
            String id = (String) identifier.get("recordId");
            String mt = (String) identifier.get("mt");

            if (StringUtils.isEmpty(tableName) || StringUtils.isEmpty(id)) {
                continue;
            }

            // 格式化ID和mt
            String formattedId = id + (StringUtils.isEmpty(mt) ? "" : ":" + mt);

            groupedByTable.computeIfAbsent(tableName, k -> new ArrayList<>())
                    .add(formattedId);
        }

        log.info("标识信息已按表名分组，共 {} 个不同表名", groupedByTable.size());

        // 2. 创建线程池
        int processors = Runtime.getRuntime().availableProcessors();
        // 使用处理器数量的2倍作为线程池大小，但不超过表的数量
        int threadPoolSize = Math.min(processors * 2, groupedByTable.size());
        log.info("创建线程池，大小: {}", threadPoolSize);

        ExecutorService executorService = Executors.newFixedThreadPool(threadPoolSize);

        // 3. 使用CompletableFuture进行并行查询
        List<CompletableFuture<List<Company>>> futures = new ArrayList<>();

        for (Map.Entry<String, List<String>> entry : groupedByTable.entrySet()) {
            String tableName = entry.getKey();
            List<String> ids = entry.getValue();

            // 分批处理，每批1000条
            int batchSize = 1000;
            for (int i = 0; i < ids.size(); i += batchSize) {
                int end = Math.min(i + batchSize, ids.size());
                List<String> batchIds = ids.subList(i, end);

                // 提交任务
                CompletableFuture<List<Company>> future = CompletableFuture.supplyAsync(() -> {
                    log.info("开始查询表 {}, 批次大小: {}", tableName, batchIds.size());
                    List<Company> result = new ArrayList<>();
                    int maxRetries = 3;
                    int retryCount = 0;

                    // 使用直接JDBC连接查询，避开ShardingSphere处理
                    Connection conn = null;
                    PreparedStatement ps = null;
                    ResultSet rs = null;

                    // 添加重试逻辑
                    while (retryCount < maxRetries) {
                        try {
                            // 构建IN子句和参数列表
                            StringBuilder placeholders = new StringBuilder();
                            for (int j = 0; j < batchIds.size(); j++) {
                                placeholders.append(j > 0 ? ",?" : "?");
                            }

                            // 获取数据源中的原始连接，使用companyDataSource而不是通过ShardingSphere
                            conn = companyDataSource.getConnectionForTable(tableName);

                            // 构建SQL语句，指定数据库名并简化查询条件
                            String sql = "SELECT * FROM zoominfo." + tableName +
                                    " WHERE id IN (" +
                                    placeholders.toString() + ")";

                            log.debug("执行SQL: {}", sql);

                            // 直接使用JDBC API
                            ps = conn.prepareStatement(sql);

                            // 设置参数 - 需要将复合ID (id:mt) 拆分，只使用数字部分
                            log.debug("准备参数，batchIds大小: {}", batchIds.size());
                            for (int j = 0; j < batchIds.size(); j++) {
                                String fullId = batchIds.get(j);
                                // 如果ID包含冒号，只取冒号前的部分（数字ID）
                                String numericId = fullId.contains(":") ? fullId.substring(0, fullId.indexOf(':')) : fullId;
                                ps.setString(j + 1, numericId);
                                if (j < 5) {
                                    log.debug("参数 {}: 原始ID={}, 处理后ID={}", j + 1, fullId, numericId);
                                }
                            }

                            // 执行查询
                            rs = ps.executeQuery();

                            // 手动映射结果
                            while (rs.next()) {
                                Company company = new Company();

                                // 获取所有字段并设置到Company对象中
                                try {
                                    company.setId(rs.getLong("id"));
                                } catch (Exception e) {
                                }
                                try {
                                    company.setCompanyName(!Objects.equals(rs.getString("COMPANY_NAME"), "\\N") ? rs.getString("COMPANY_NAME") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setPrimaryCity(!Objects.equals(rs.getString("COMPANY_CITY"), "\\N") ? rs.getString("COMPANY_CITY") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setPrimaryState(!Objects.equals(rs.getString("COMPANY_STATE"), "\\N") ? rs.getString("COMPANY_STATE") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setIndustry(!Objects.equals(rs.getString("COMPANY_INDUSTRY"), "\\N") ? rs.getString("COMPANY_INDUSTRY") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setSubIndustry(!Objects.equals(rs.getString("PRIMARY_INDUSTRY"), "\\N") ? rs.getString("PRIMARY_INDUSTRY") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setEmployee(!Objects.equals(rs.getString("COMPANY_EMPLOYEE_COUNT"), "\\N") ? rs.getString("COMPANY_EMPLOYEE_COUNT") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setReveneu(!Objects.equals(rs.getString("COMPANY_REVENUE"), "\\N") ? rs.getString("COMPANY_REVENUE") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setFirstName(!Objects.equals(rs.getString("FIRST_NAME"), "\\N") ? rs.getString("FIRST_NAME") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setLastName(!Objects.equals(rs.getString("LAST_NAME"), "\\N") ? rs.getString("LAST_NAME") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setTitle(!Objects.equals(rs.getString("JOB_TITLE"), "\\N") ? rs.getString("JOB_TITLE") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setPhone(!Objects.equals(rs.getString("COMPANY_PHONE"), "\\N") ? rs.getString("COMPANY_PHONE") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setMobile(!Objects.equals(rs.getString("MOBILE_PHONE"), "\\N") ? rs.getString("MOBILE_PHONE") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setEmail(!Objects.equals(rs.getString("BUSINESS_EMAIL"), "\\N") ? rs.getString("BUSINESS_EMAIL") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setMailingAddress(!Objects.equals(rs.getString("COMPANY_ADDRESS"), "\\N") ? rs.getString("COMPANY_ADDRESS") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setZipCode(!Objects.equals(rs.getString("COMPANY_ZIP"), "\\N") ? rs.getString("COMPANY_ZIP") : null);
                                } catch (Exception e) {
                                }
                                try {
                                    company.setWebAddress(!Objects.equals(rs.getString("COMPANY_LINKEDIN_URL"), "\\N") ? rs.getString("COMPANY_LINKEDIN_URL") : null);
                                } catch (Exception e) {
                                }

                                result.add(company);
                            }

                            log.info("表 {} 批量JDBC查询完成，返回 {} 条记录", tableName, result.size());

                            // 后处理：将ES返回的mt信息与查询结果关联
                            if (!result.isEmpty()) {
                                // 构建ID到mt的映射
                                Map<String, String> idToMtMap = new HashMap<>();
                                for (String fullId : batchIds) {
                                    if (fullId.contains(":")) {
                                        String idPart = fullId.substring(0, fullId.indexOf(':'));
                                        String mtPart = fullId.substring(fullId.indexOf(':') + 1);
                                        idToMtMap.put(idPart, mtPart);
                                    }
                                }

                                // 遍历查询结果，为每个没有mt值的记录设置mt
                                for (Company company : result) {
                                    if (company.getMt() == null && idToMtMap.containsKey(String.valueOf(company.getId()))) {
                                        company.setMt(idToMtMap.get(String.valueOf(company.getId())));
                                    }
                                }
                            }

                            // 查询成功，跳出重试循环
                            break;
                        } catch (Exception e) {
                            retryCount++;
                            log.error("原生JDBC查询表 {} 批次数据失败 (尝试 {}/{})", tableName, retryCount, maxRetries, e);

                            if (retryCount >= maxRetries) {
                                log.error("已达到最大重试次数，放弃查询表 {}", tableName);
                            } else {
                                // 重试前暂停一下
                                try {
                                    Thread.sleep(1000 * retryCount); // 每次重试增加等待时间
                                } catch (InterruptedException ie) {
                                    Thread.currentThread().interrupt();
                                }
                            }
                        } finally {
                            // 关闭资源
                            try {
                                if (rs != null) rs.close();
                                if (ps != null) ps.close();
                                if (conn != null) conn.close();
                            } catch (Exception e) {
                                log.error("关闭JDBC资源失败", e);
                            }
                        }
                    }

                    return result;
                }, executorService);

                futures.add(future);
            }
        }

        // 4. 合并所有查询结果
        List<Company> enterprises = new ArrayList<>();

        try {
            // 等待所有futures完成
            CompletableFuture<Void> allFutures = CompletableFuture.allOf(
                    futures.toArray(new CompletableFuture[0])
            );

            // 等待所有异步任务完成，设置最大等待时间为5分钟
            allFutures.get(5, TimeUnit.MINUTES);

            // 收集所有结果
            for (CompletableFuture<List<Company>> future : futures) {
                try {
                    List<Company> result = future.get();
                    enterprises.addAll(result);
                } catch (Exception e) {
                    log.error("获取查询结果失败", e);
                }
            }

        } catch (Exception e) {
            log.error("等待并行查询完成时出错", e);
        } finally {
            // 关闭线程池
            executorService.shutdown();
            try {
                // 等待线程池关闭，最多等待30秒
                if (!executorService.awaitTermination(30, TimeUnit.SECONDS)) {
                    log.warn("线程池未能在30秒内正常关闭，将强制关闭");
                    executorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                log.error("关闭线程池时被中断", e);
                Thread.currentThread().interrupt(); // 重置中断标志
                executorService.shutdownNow();
            }
        }

        log.info("第二阶段多线程批量查询完成，共获取 {} 条数据", enterprises.size());
        return enterprises;
    }
} 